package com.hitqz.robot.biz.websocket.endpoint;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;

import java.io.IOException;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

//@ServerEndpoint(value = "/websocket/publish/{clientId}/{topic}")
//@Component
@Deprecated
@Slf4j
public class PublishingEndpoint {
	// ObjectMapper for JSON parsing
	private static final ObjectMapper objectMapper = new ObjectMapper();
	// 用于按topic存储clientId到消息的映射
	private static final Map<String, Map<String, String>> topicClientMessages = new ConcurrentHashMap<>();
	// 用于存储所有打开的WebSocket sessions
	private static final Set<Session> sessions = new CopyOnWriteArraySet<>();

	@OnOpen
	public void onOpen(@PathParam(value = "clientId") String clientId,@PathParam(value = "topic") String topic, Session session) {
		sessions.add(session);
		log.debug("[ws][publish]{},topic:{},session-id:{} connected.current size :{}", clientId,topic,session.getId(),sessions.size());
		
	}

	// 接收客户端JSON消息推送,按topic汇总后广播
	@OnMessage
	public void onMessage(@PathParam(value = "clientId") String clientId,@PathParam(value = "topic") String topic,Session session,String message) {
		try {
			// 判断message是否是json格式
			JsonNode jsonNode = objectMapper.readTree(message);
			if (!jsonNode.isObject()) {
				throw new IllegalArgumentException("Invalid JSON message");
			}
			// 将消息添加到对应的topic映射中，根据clientId去重
			topicClientMessages.computeIfAbsent(topic, k -> new ConcurrentHashMap<>()).put(clientId, jsonNode.toString());

			// 准备广播消息
			// fixme 这边是收到消息后立即广播，可以根据需要调整广播时机
			broadcastMessageByTopic(topic);
		} catch (Exception e) {
			log.error("[ws][publish]Failed to parse incoming message: {}", e.getMessage());
			// 可以根据需要发送错误消息给客户端或关闭会话
			sendErrorAndCloseSession(clientId, session,"parse message error");
		}
	}

	@OnClose
	public void onClose(@PathParam(value = "clientId") String clientId,@PathParam(value = "topic") String topic,Session session) {
		// 移除session
		sessions.remove(session);
		//清除缓存
		topicClientMessages.getOrDefault(topic, Collections.emptyMap()).remove(clientId);
		//进一步去除topic缓存
		topicClientMessages.entrySet().removeIf(entry -> entry.getValue().isEmpty());
		log.debug("[ws][publish]{},topic:{},session-id:{} disconnected.current size :{}", clientId,topic,session.getId(),sessions.size());
	}

	@OnError
	public void onError(@PathParam(value = "clientId") String clientId,Session session,Throwable error) {
		log.error("[ws][publish]{},session-id:{} error:{}", clientId,session.getId(),error.getMessage());
	}

	private void broadcastMessageByTopic(String topic) {
		Map<String, String> clientMessages = topicClientMessages.getOrDefault(topic, Collections.emptyMap());
		if (clientMessages.isEmpty()) {
			return;
		}
		// 创建广播的消息内容，包含所有去重后的消息
		String broadcastMessage = createBroadcastMessage(clientMessages);
		// 广播消息给所有订阅的会话
		SubscriptionEndpoint.broadcast(topic, broadcastMessage);
	}

	private String createBroadcastMessage( Map<String, String> clientMessages) {
		try {
			// 将去重后的消息转换成JSONArray
			ArrayNode messagesArray = objectMapper.createArrayNode();
			//按key的顺序去添加
			clientMessages.entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(entry -> {
				try {
					messagesArray.add(objectMapper.readTree(entry.getValue()));
				} catch (IOException e) {
					log.error("[ws][publish]Failed to parse message: {}", e.getMessage());
				}
			});
			//转成不带换行符的json字符串
			return objectMapper.writeValueAsString(messagesArray);
		} catch (Exception e) {
			log.error("[ws][publish]Failed to create broadcast message: {}", e.getMessage());
			return "{\"error\":\"Failed to create broadcast message\"}";
		}
	}


	private void sendErrorAndCloseSession(String clientId,Session session, String errorMessage) {
		try {
			session.getBasicRemote().sendText("{\"error\":\"" + errorMessage + "\"}");
		} catch (IOException e) {
			// 处理发送错误消息时的异常
			log.error("[ws][publish]send to {} error message failed:{}",clientId,e.getMessage());
		}finally {
			try {
				// 在发送错误信息后关闭会话
				session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT, errorMessage));
			} catch (IOException e) {
				// 处理关闭session时的异常
				log.error("[ws][publish]close {}, session-id:{} failed:{}",clientId,session.getId(),e.getMessage());
			}
		}
	}

	// 定时向所有会话发送topic消息
	@Scheduled(fixedRate = 1000)
	public void ping() {
		topicClientMessages.keySet().forEach(this::broadcastMessageByTopic);
	}



}
